package com.soushin.mvpdemo.module.rxjava.operator;

import com.blankj.ALog;
import com.zhouyou.http.EasyHttp;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;

import io.reactivex.Completable;
import io.reactivex.CompletableObserver;
import io.reactivex.Flowable;
import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.Observer;
import io.reactivex.Single;
import io.reactivex.SingleObserver;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.annotations.NonNull;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.BiFunction;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.functions.Predicate;
import io.reactivex.schedulers.Schedulers;
import io.reactivex.subjects.AsyncSubject;
import io.reactivex.subjects.BehaviorSubject;
import io.reactivex.subjects.PublishSubject;
import com.soushin.mvpdemo.Utils.TimeUtil;
import com.soushin.mvpdemo.base.BasePresenter;

/**
 * 操作符
 *
 * @author SouShin
 * @time 2018/11/24 14:21
 */
public class OperatorPresenter extends BasePresenter<OperatorView> {
    private Disposable mDisposable;

    @Override
    public void subscribe(OperatorView mvpView) {
        super.subscribe(mvpView);
    }

    public void doSomeThing() {
        getMvpView().setResult("");
        switch (getMvpView().getClickPosition()) {
            case 0:
                goCreate();
                break;
            case 1:
                goZip();
                break;
            case 2:
                goMap();
                break;
            case 3:
                goFlatmap();
                break;
            case 4:
                goConcatmap();
                break;
            case 5:
                goDoOnNext();
                break;
            case 6:
                goFilter();
                break;
            case 7:
                goSkip();
                break;
            case 8:
                goTake();
                break;
            case 9:
                goTimer();
                break;
            case 10:
                goInterval();
                break;
            case 11:
                goJust();
                break;
            case 12:
                goSingle();
                break;
            case 13:
                goConcat();
                break;
            case 14:
                goDistinct();
                break;
            case 15:
                goBuffer();
                break;
            case 16:
                goDebounce();
                break;
            case 17:
                goDefer();
                break;
            case 18:
                goLast();
                break;
            case 19:
                goMerge();
                break;
            case 20:
                goReduce();
                break;
            case 21:
                goScan();
                break;
            case 22:
                goWindow();
                break;
            case 23:
                goPublishSubject();
                break;
            case 24:
                goAsyncSubject();
                break;
            case 25:
                goBehaviorSubject();
                break;
            case 26:
                goCompletable();
                break;
            case 27:
                goFlowable();
                break;
        }
    }


    /**
     * create
     * 最常见的操作符，用于生产一个发射对象
     */
    private void goCreate() {

        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                appendMessage("Observable emit 1" + "\n");
                ALog.e("Observable emit 1" + "\n");
                e.onNext(1);
                appendMessage("Observable emit 2" + "\n");
                ALog.e("Observable emit 2" + "\n");
                e.onNext(2);
                appendMessage("Observable emit 3" + "\n");
                ALog.e("Observable emit 3" + "\n");
                e.onNext(3);
                e.onComplete();
                appendMessage("Observable emit 4" + "\n");
                ALog.e("Observable emit 4" + "\n");
                e.onNext(4);
            }
        }).subscribe(new Observer<Integer>() {
            private int i;
            private Disposable mDisposable;

            @Override
            public void onSubscribe(@NonNull Disposable d) {
                appendMessage("onSubscribe : " + d.isDisposed() + "\n");
                ALog.e("onSubscribe : " + d.isDisposed() + "\n");
                mDisposable = d;
            }

            @Override
            public void onNext(@NonNull Integer integer) {
                appendMessage("onNext : value : " + integer + "\n");
                ALog.e("onNext : value : " + integer + "\n");
                i++;
                if (i == 2) {
                    // 在RxJava 2.x 中，新增的Disposable可以做到切断的操作，让Observer观察者不再接收上游事件
                    mDisposable.dispose();
                    appendMessage("onNext : isDisposable : " + mDisposable.isDisposed() + "\n");
                    ALog.e("onNext : isDisposable : " + mDisposable.isDisposed() + "\n");
                }
            }

            @Override
            public void onError(@NonNull Throwable e) {
                appendMessage("onError : value : " + e.getMessage() + "\n");
                ALog.e("onError : value : " + e.getMessage() + "\n");
            }

            @Override
            public void onComplete() {
                appendMessage("onComplete" + "\n");
                ALog.e("onComplete" + "\n");
            }
        });
    }

    /**
     * zip 合并事件专用
     * 分别从两个上游事件中各取出一个组合
     * 一个事件只能被使用一次，顺序严格按照事件发送的顺序
     * 最终下游事件收到的是和上游事件最少的数目相同（必须两两配对，多余的舍弃)
     */
    private void goZip() {
        mDisposable = Observable.zip(getStringObservable(), getIntegerObservable(), new BiFunction<String, Integer, String>() {
            @Override
            public String apply(@NonNull String s, @NonNull Integer integer) throws Exception {
                ALog.e("zip线程",Thread.currentThread().getName());
                return s + integer;
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(@NonNull String s) throws Exception {
                ALog.e("subscribe线程", Thread.currentThread().getName());
                appendMessage("zip : accept : " + s + "\n");
                ALog.e("zip : accept : " + s + "\n");
            }
        }, new Consumer<Throwable>() {
            @Override
            public void accept(Throwable throwable) throws Exception {
                ALog.e("subscribe异常",throwable.getMessage());
            }
        });
    }

    private Observable<String> getStringObservable() {
        return Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<String> e) throws Exception {
                if (!e.isDisposed()) {
                    e.onNext("A");
                    appendMessage("String emit : A \n");
                    ALog.e("String emit : A \n");
                    e.onNext("B");
                    appendMessage("String emit : B \n");
                    ALog.e("String emit : B \n");
                    e.onNext("C");
                    appendMessage("String emit : C \n");
                    ALog.e("String emit : C \n");
                }
            }
        });
    }

    private Observable<Integer> getIntegerObservable() {
        return Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
                if (!e.isDisposed()) {
                    e.onNext(1);
                    appendMessage("Integer emit : 1 \n");
                    ALog.e("Integer emit : 1 \n");
                    e.onNext(2);
                    appendMessage("Integer emit : 2 \n");
                    ALog.e("Integer emit : 2 \n");
                    e.onNext(3);
                    appendMessage("Integer emit : 3 \n");
                    ALog.e("Integer emit : 3 \n");
                    e.onNext(4);
                    appendMessage("Integer emit : 4 \n");
                    ALog.e("Integer emit : 4 \n");
                    e.onNext(5);
                    appendMessage("Integer emit : 5 \n");
                    ALog.e("Integer emit : 5 \n");
                }
            }
        });
    }

    /**
     * map
     * 基本是RxJava 最简单的操作符了
     * 作用是对上游发送的每一个事件应用一个函数，使得每一个事件都按照指定的函数去变化
     */
    private void goMap() {
        mDisposable = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
                e.onNext(1);
                e.onNext(2);
                e.onNext(3);
            }
        }).map(new Function<Integer, String>() {
            @Override
            public String apply(@NonNull Integer integer) throws Exception {
                return "This is result " + integer;
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(@NonNull String s) throws Exception {
                appendMessage("accept : " + s + "\n");
                ALog.e("accept : " + s + "\n");
            }
        });
    }

    /**
     * FlatMap将一个发送事件的上游Observable变换成多个发送事件的Observables，
     * 然后将它们发射的时间合并后放进一个单独的Observable里
     * [需要注意]flatMap并不保证事件的顺序
     */
    private void goFlatmap() {
        mDisposable = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
                e.onNext(1);
                e.onNext(2);
                e.onNext(3);
            }
        }).flatMap(new Function<Integer, ObservableSource<String>>() {
            @Override
            public ObservableSource<String> apply(@NonNull Integer integer) throws Exception {
                List<String> list = new ArrayList<>();
                for (int i = 0; i < 3; i++) {
                    list.add("I am value " + integer);
                }
                int delayTime = (int) (1 + Math.random() * 10);
                return Observable.fromIterable(list).delay(delayTime, TimeUnit.MILLISECONDS);
            }
        }).subscribeOn(Schedulers.newThread())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(@NonNull String s) throws Exception {
                        ALog.e("flatMap : accept : " + s + "\n");
                        appendMessage("flatMap : accept : " + s + "\n");
                    }
                });
    }

    /**
     * concatMap
     * concatMap作用和flatMap几乎一模一样，唯一的区别是它能保证事件的顺序
     */
    private void goConcatmap() {
        mDisposable = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
                e.onNext(1);
                e.onNext(2);
                e.onNext(3);
            }
        }).concatMap(new Function<Integer, ObservableSource<String>>() {
            @Override
            public ObservableSource<String> apply(@NonNull Integer integer) throws Exception {
                List<String> list = new ArrayList<>();
                for (int i = 0; i < 3; i++) {
                    list.add("I am value " + integer);
                }
                int delayTime = (int) (1 + Math.random() * 10);
                return Observable.fromIterable(list).delay(delayTime, TimeUnit.MILLISECONDS);
            }
        }).subscribeOn(Schedulers.newThread())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(@NonNull String s) throws Exception {
                        ALog.e("concatMap : accept : " + s + "\n");
                        appendMessage("concatMap : accept : " + s + "\n");
                    }
                });
    }

    /**
     * doOnNext
     * 让订阅者在接收到数据前干点事情的操作符
     */
    private void goDoOnNext() {
        mDisposable = Observable.just(1, 2, 3, 4)
                .doOnNext(new Consumer<Integer>() {
                    @Override
                    public void accept(@NonNull Integer integer) throws Exception {
                        appendMessage("doOnNext 保存 " + integer + "成功" + "\n");
                        ALog.e("doOnNext 保存 " + integer + "成功" + "\n");
                    }
                }).subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(@NonNull Integer integer) throws Exception {
                        appendMessage("doOnNext :" + integer + "\n");
                        ALog.e("doOnNext :" + integer + "\n");
                    }
                });
    }

    /**
     * filter
     * 过滤操作符，取想要的值
     */
    private void goFilter() {
        mDisposable = Observable.just(1, 20, 65, -5, 7, 19)
                .filter(new Predicate<Integer>() {
                    @Override
                    public boolean test(@NonNull Integer integer) throws Exception {
                        return integer >= 10;
                    }
                }).subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(@NonNull Integer integer) throws Exception {
                        appendMessage("filter : " + integer + "\n");
                        ALog.e("filter : " + integer + "\n");
                    }
                });
    }

    /**
     * skip
     * 接受一个long型参数，代表跳过多少个数目的事件再开始接收
     */
    private void goSkip() {

        mDisposable = Observable.just(1, 2, 3, 4, 5)
                .skip(2)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(@NonNull Integer integer) throws Exception {
                        appendMessage("skip : " + integer + "\n");
                        ALog.e("skip : " + integer + "\n");
                    }
                });
    }

    /**
     * take
     * 用于指定订阅者最多收到多少数据
     */
    private void goTake() {
        mDisposable = Flowable.fromArray(1, 2, 3, 4, 5)
                .take(2)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(@NonNull Integer integer) throws Exception {
                        appendMessage("take : " + integer + "\n");
                        ALog.e("accept: take : " + integer + "\n");
                    }
                });
    }

    /**
     * timer
     * 定时任务
     */
    private void goTimer() {

        appendMessage("timer start : " + TimeUtil.getNowStrTime() + "\n");
        ALog.e("timer start : " + TimeUtil.getNowStrTime() + "\n");
        mDisposable = Observable.timer(2, TimeUnit.SECONDS)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<Long>() {
                    @Override
                    public void accept(@NonNull Long aLong) throws Exception {
                        appendMessage("timer :" + aLong + " at " + TimeUtil.getNowStrTime() + "\n");
                        ALog.e("timer :" + aLong + " at " + TimeUtil.getNowStrTime() + "\n");
                    }
                });
    }

    /**
     * interval
     * 间隔执行操作，默认在新线程
     */
    private void goInterval() {
        appendMessage("interval start : " + TimeUtil.getNowStrTime() + "\n");
        ALog.e("interval start : " + TimeUtil.getNowStrTime() + "\n");
        mDisposable = Observable.interval(3, 2, TimeUnit.SECONDS)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread()) // 由于interval默认在新线程，所以我们应该切回主线程
                .subscribe(new Consumer<Long>() {
                    @Override
                    public void accept(@NonNull Long aLong) throws Exception {
                        appendMessage("interval :" + aLong + " at " + TimeUtil.getNowStrTime() + "\n");
                        ALog.e("interval :" + aLong + " at " + TimeUtil.getNowStrTime() + "\n");
                    }
                });
    }

    /**
     * just
     * 相当于在create()发送事件
     */
    private void goJust() {

        mDisposable = Observable.just("1", "2", "3")
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(@NonNull String s) throws Exception {
                        appendMessage("accept : onNext : " + s + "\n");
                        ALog.e("accept : onNext : " + s + "\n");
                    }
                });
    }

    /**
     * single
     * 顾名思义，Single只会接收一个参数
     * 而SingleObserver只会调用onError或者onSuccess
     */
    private void goSingle() {
        Single.just(new Random().nextInt())
                .subscribe(new SingleObserver<Integer>() {
                    @Override
                    public void onSubscribe(@NonNull Disposable d) {
                    }

                    @Override
                    public void onSuccess(@NonNull Integer integer) {
                        appendMessage("single : onSuccess : " + integer + "\n");
                        ALog.e("single : onSuccess : " + integer + "\n");
                    }

                    @Override
                    public void onError(@NonNull Throwable e) {
                        appendMessage("single : onError : " + e.getMessage() + "\n");
                        ALog.e("single : onError : " + e.getMessage() + "\n");
                    }
                });
    }

    /**
     * concat
     * 连接操作符，可接受Observable的可变参数，或者Observable的集合
     */
    private void goConcat() {
        mDisposable = Observable.concat(Observable.just(1, 2, 3), Observable.just(4, 5, 6))
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(@NonNull Integer integer) throws Exception {
                        appendMessage("concat : " + integer + "\n");
                        ALog.e("concat : " + integer + "\n");
                    }
                });
    }

    /**
     * distinct
     * 去重操作符，其实就是简单的去重
     */
    private void goDistinct() {
        mDisposable = Observable.just(1, 1, 1, 2, 2, 3, 4, 5)
                .distinct()
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(@NonNull Integer integer) throws Exception {
                        appendMessage("distinct : " + integer + "\n");
                        ALog.e("distinct : " + integer + "\n");
                    }
                });
    }

    /**
     * buffer
     * buffer(count, skip)` 从定义就差不多能看出作用了，
     * 将 observable 中的数据按 skip（步长）分成最长不超过 count 的 buffer，然后生成一个 observable
     */
    private void goBuffer() {
        mDisposable = Observable.just(1, 2, 3, 4, 5)
                .buffer(3, 2)
                .subscribe(new Consumer<List<Integer>>() {
                    @Override
                    public void accept(@NonNull List<Integer> integers) throws Exception {
                        appendMessage("buffer size : " + integers.size() + "\n");
                        ALog.e("buffer size : " + integers.size() + "\n");
                        appendMessage("buffer value : ");
                        ALog.e("buffer value : ");
                        for (Integer i : integers) {
                            appendMessage(i + "");
                            ALog.e(i + "");
                        }
                        appendMessage("\n");
                    }
                });
    }

    /**
     * debounce
     * 过滤掉发射速率过快的数据项
     */
    private void goDebounce() {
        mDisposable = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<Integer> emitter) throws Exception {
                // send events with simulated time wait
                emitter.onNext(1); // skip
                Thread.sleep(400);
                emitter.onNext(2); // deliver
                Thread.sleep(505);
                emitter.onNext(3); // skip
                Thread.sleep(100);
                emitter.onNext(4); // deliver
                Thread.sleep(605);
                emitter.onNext(5); // deliver
                Thread.sleep(510);
                emitter.onComplete();
            }
        }).debounce(500, TimeUnit.MILLISECONDS)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(@NonNull Integer integer) throws Exception {
                        appendMessage("debounce :" + integer + "\n");
                        ALog.e("debounce :" + integer + "\n");
                    }
                });
    }

    /**
     * defer
     * 简单的说就是每次订阅都会创建一个新的Observable
     * 并且如果该Observable没有被订阅，就不会生成新的Observable
     */
    private void goDefer() {

        Observable<Integer> observable = Observable.defer(new Callable<ObservableSource<Integer>>() {
            @Override
            public ObservableSource<Integer> call() throws Exception {
                return Observable.just(1, 2, 3);
            }
        });

        observable.subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {

            }

            @Override
            public void onNext(@NonNull Integer integer) {
                appendMessage("defer : " + integer + "\n");
                ALog.e("defer : " + integer + "\n");
            }

            @Override
            public void onError(@NonNull Throwable e) {
                appendMessage("defer : onError : " + e.getMessage() + "\n");
                ALog.e("defer : onError : " + e.getMessage() + "\n");
            }

            @Override
            public void onComplete() {
                appendMessage("defer : onComplete\n");
                ALog.e("defer : onComplete\n");
            }
        });
    }

    /**
     * last
     * 取出最后一个值，参数是没有值的时候的默认值
     */
    private void goLast() {
        mDisposable = Observable.just(1, 2, 3)
                .last(4)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(@NonNull Integer integer) throws Exception {
                        appendMessage("last : " + integer + "\n");
                        ALog.e("last : " + integer + "\n");
                    }
                });
    }

    /**
     * merge
     * 将多个Observable合起来，接受可变参数，也支持使用迭代器集合
     */
    private void goMerge() {
        mDisposable = Observable.merge(Observable.just(1, 2), Observable.just(3, 4, 5))
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(@NonNull Integer integer) throws Exception {
                        appendMessage("merge :" + integer + "\n");
                        ALog.e("accept: merge :" + integer + "\n");
                    }
                });
    }

    /**
     * reduce
     * 就是一次用一个方法处理一个值，可以有一个seed作为初始值
     */
    private void goReduce() {
        mDisposable = Observable.just(1, 2, 3)
                .reduce(new BiFunction<Integer, Integer, Integer>() {
                    @Override
                    public Integer apply(@NonNull Integer integer, @NonNull Integer integer2) throws Exception {
                        return integer + integer2;
                    }
                }).subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(@NonNull Integer integer) throws Exception {
                        appendMessage("reduce : " + integer + "\n");
                        ALog.e("accept: reduce : " + integer + "\n");
                    }
                });
    }

    /**
     * scan
     * 和上面的reduce差不多，区别在于reduce()只输出结果，而scan()会将过程中每一个结果输出
     */
    private void goScan() {
        mDisposable = Observable.just(1, 2, 3)
                .scan(new BiFunction<Integer, Integer, Integer>() {
                    @Override
                    public Integer apply(@NonNull Integer integer, @NonNull Integer integer2) throws Exception {
                        return integer + integer2;
                    }
                }).subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(@NonNull Integer integer) throws Exception {
                        appendMessage("scan " + integer + "\n");
                        ALog.e("accept: scan " + integer + "\n");
                    }
                });
    }

    /**
     * window
     * 按照时间划分窗口，将数据发送给不同的Observable
     */
    private void goWindow() {
        mDisposable = Observable.interval(1, TimeUnit.SECONDS) // 间隔一秒发一次
                .take(15) // 最多接收15个
                .window(3, TimeUnit.SECONDS)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<Observable<Long>>() {
                    @Override
                    public void accept(@NonNull Observable<Long> longObservable) throws Exception {
                        appendMessage("Sub Divide begin...\n");
                        ALog.e("Sub Divide begin...\n");
                        longObservable.subscribeOn(Schedulers.io())
                                .observeOn(AndroidSchedulers.mainThread())
                                .subscribe(new Consumer<Long>() {
                                    @Override
                                    public void accept(@NonNull Long aLong) throws Exception {
                                        appendMessage("Next:" + aLong + "\n");
                                        ALog.e("Next:" + aLong + "\n");
                                    }
                                });
                    }
                });
    }

    /**
     * onNext() 会通知每个观察者，仅此而已
     */
    private void goPublishSubject() {
        PublishSubject<Integer> publishSubject = PublishSubject.create();

        publishSubject.subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                appendMessage("First onSubscribe :" + d.isDisposed() + "\n");
                ALog.e("First onSubscribe :" + d.isDisposed() + "\n");
            }

            @Override
            public void onNext(@NonNull Integer integer) {
                appendMessage("First onNext value :" + integer + "\n");
                ALog.e("First onNext value :" + integer + "\n");
            }

            @Override
            public void onError(@NonNull Throwable e) {
                appendMessage("First onError:" + e.getMessage() + "\n");
                ALog.e("First onError:" + e.getMessage() + "\n");
            }

            @Override
            public void onComplete() {
                appendMessage("First onComplete!\n");
                ALog.e("First onComplete!\n");
            }
        });

        publishSubject.onNext(1);
        publishSubject.onNext(2);
        publishSubject.onNext(3);

        publishSubject.subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                appendMessage("Second onSubscribe :" + d.isDisposed() + "\n");
                ALog.e("Second onSubscribe :" + d.isDisposed() + "\n");
            }

            @Override
            public void onNext(@NonNull Integer integer) {
                appendMessage("Second onNext value :" + integer + "\n");
                ALog.e("Second onNext value :" + integer + "\n");
            }

            @Override
            public void onError(@NonNull Throwable e) {
                appendMessage("Second onError:" + e.getMessage() + "\n");
                ALog.e("Second onError:" + e.getMessage() + "\n");
            }

            @Override
            public void onComplete() {
                appendMessage("Second onComplete!\n");
                ALog.e("Second onComplete!\n");
            }
        });
        publishSubject.onNext(4);
        publishSubject.onNext(5);
        publishSubject.onComplete();
    }

    /**
     * AsyncSubject
     * 在 调用 onComplete() 之前，除了 subscribe() 其它的操作都会被缓存，
     * 在调用 onComplete() 之后只有最后一个 onNext() 会生效
     */
    private void goAsyncSubject() {
        AsyncSubject<Integer> asyncSubject = AsyncSubject.create();

        asyncSubject.subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                appendMessage("First onSubscribe :" + d.isDisposed() + "\n");
                ALog.e("First onSubscribe :" + d.isDisposed() + "\n");
            }

            @Override
            public void onNext(@NonNull Integer integer) {
                appendMessage("First onNext value :" + integer + "\n");
                ALog.e("First onNext value :" + integer + "\n");
            }

            @Override
            public void onError(@NonNull Throwable e) {
                appendMessage("First onError:" + e.getMessage() + "\n");
                ALog.e("First onError:" + e.getMessage() + "\n");
            }

            @Override
            public void onComplete() {
                appendMessage("First onComplete!\n");
                ALog.e("First onComplete!\n");
            }
        });

        asyncSubject.onNext(1);
        asyncSubject.onNext(2);
        asyncSubject.onNext(3);

        asyncSubject.subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                appendMessage("Second onSubscribe :" + d.isDisposed() + "\n");
                ALog.e("Second onSubscribe :" + d.isDisposed() + "\n");
            }

            @Override
            public void onNext(@NonNull Integer integer) {
                appendMessage("Second onNext value :" + integer + "\n");
                ALog.e("Second onNext value :" + integer + "\n");
            }

            @Override
            public void onError(@NonNull Throwable e) {
                appendMessage("Second onError:" + e.getMessage() + "\n");
                ALog.e("Second onError:" + e.getMessage() + "\n");
            }

            @Override
            public void onComplete() {
                appendMessage("Second onComplete!\n");
                ALog.e("Second onComplete!\n");
            }
        });

        asyncSubject.onNext(4);
        asyncSubject.onNext(5);
        asyncSubject.onComplete();
    }

    /**
     * BehaviorSubject
     * BehaviorSubject 的最后一次 onNext() 操作会被缓存，然后在 subscribe() 后立刻推给新注册的 Observer
     */
    private void goBehaviorSubject() {
        BehaviorSubject<Integer> behaviorSubject = BehaviorSubject.create();

        behaviorSubject.subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                appendMessage("First onSubscribe :" + d.isDisposed() + "\n");
                ALog.e("First onSubscribe :" + d.isDisposed() + "\n");
            }

            @Override
            public void onNext(@NonNull Integer integer) {
                appendMessage("First onNext value :" + integer + "\n");
                ALog.e("First onNext value :" + integer + "\n");
            }

            @Override
            public void onError(@NonNull Throwable e) {
                appendMessage("First onError:" + e.getMessage() + "\n");
                ALog.e("First onError:" + e.getMessage() + "\n");
            }

            @Override
            public void onComplete() {
                appendMessage("First onComplete!\n");
                ALog.e("First onComplete!\n");
            }
        });

        behaviorSubject.onNext(1);
        behaviorSubject.onNext(2);
        behaviorSubject.onNext(3);

        behaviorSubject.subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                appendMessage("Second onSubscribe :" + d.isDisposed() + "\n");
                ALog.e("Second onSubscribe :" + d.isDisposed() + "\n");
            }

            @Override
            public void onNext(@NonNull Integer integer) {
                appendMessage("Second onNext value :" + integer + "\n");
                ALog.e("Second onNext value :" + integer + "\n");
            }

            @Override
            public void onError(@NonNull Throwable e) {
                appendMessage("Second onError:" + e.getMessage() + "\n");
                ALog.e("Second onError:" + e.getMessage() + "\n");
            }

            @Override
            public void onComplete() {
                appendMessage("Second onComplete!\n");
                ALog.e("Second onComplete!\n");
            }
        });

        behaviorSubject.onNext(4);
        behaviorSubject.onNext(5);
        behaviorSubject.onComplete();
    }

    /**
     * Completable
     * 只关心结果，也就是说 Completable 是没有 onNext 的，要么成功要么出错，不关心过程，在 subscribe 后的某个时间点返回结果
     */
    private void goCompletable() {
        Completable.timer(1, TimeUnit.SECONDS)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new CompletableObserver() {
                    @Override
                    public void onSubscribe(@NonNull Disposable d) {
                        appendMessage("onSubscribe : d :" + d.isDisposed() + "\n");
                        ALog.e("onSubscribe : d :" + d.isDisposed() + "\n");
                    }

                    @Override
                    public void onComplete() {
                        appendMessage("onComplete\n");
                        ALog.e("onComplete\n");
                    }

                    @Override
                    public void onError(@NonNull Throwable e) {
                        appendMessage("onError :" + e.getMessage() + "\n");
                        ALog.e("onError :" + e.getMessage() + "\n");
                    }
                });
    }

    /**
     * flowable
     * 专用于解决背压问题 (发射器发射过快,下游来不及处理,会通知上游放慢发射速度)
     */
    private void goFlowable() {
        mDisposable = Flowable.just(1, 2, 3, 4)
                .reduce(100, new BiFunction<Integer, Integer, Integer>() {
                    @Override
                    public Integer apply(@NonNull Integer integer, @NonNull Integer integer2) throws Exception {
                        return integer + integer2;
                    }
                }).subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(@NonNull Integer integer) throws Exception {
                        appendMessage("Flowable :" + integer + "\n");
                        ALog.e("Flowable :" + integer + "\n");
                    }
                });
    }

    private void appendMessage(String content) {
        getMvpView().getResultTextView().append(content);
    }

    @Override
    public void unsubscribe() {
        super.unsubscribe();
        EasyHttp.cancelSubscription(mDisposable);
    }
}
